package com.neusoft.bd.myflink.source;

import com.neusoft.bd.myflink.entity.WaterSensor;
import jdk.nashorn.internal.codegen.types.Type;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;

import java.util.Random;

public class WaterSensorSource {
    public final static Random r = new Random(System.currentTimeMillis());

    /**
     * 每秒两条水位传感器数据
     *
     * @return DataGeneratorSource<WaterSensor>
     */
    public static DataGeneratorSource<WaterSensor> waterSensorDataStreamSource() {
        GeneratorFunction<Long, WaterSensor> generatorFunction = index -> new WaterSensor((index % 2 + ""),
                System.currentTimeMillis(), (r.nextInt(10) + 90)
        );
        long numberOfRecords = 2;

        DataGeneratorSource<WaterSensor> source =
                new DataGeneratorSource<WaterSensor>(generatorFunction,
                        Long.MAX_VALUE, RateLimiterStrategy.perSecond(numberOfRecords),
                        PojoTypeInfo.of(WaterSensor.class));


        return source;
    }

}
